package com.mmc.delay.api.impl;


import com.mmc.delay.api.DelayFacade;
import com.mmc.delay.api.JobHandle;
import com.mmc.delay.bean.DeadJob;
import com.mmc.delay.bean.Job;
import com.mmc.delay.bean.JobParam;
import com.mmc.delay.bean.TopicAndId;
import com.mmc.delay.config.DelayCoreConfig;
import com.mmc.delay.container.DeadLetterSet;
import com.mmc.delay.container.DelayBucket;
import com.mmc.delay.container.JobPool;
import com.mmc.delay.container.ReadyQueue;
import com.mmc.delay.exception.DelayException;
import com.mmc.delay.helper.RedisLockHelper;
import com.mmc.delay.interceptor.SendInterceptor;
import com.mmc.delay.timer.DelayMoveToReadyTimer;
import com.mmc.delay.timer.ReadyQueueTimer;
import com.mmc.delay.util.LockUtil;
import com.mmc.delay.util.SnowFlakeShortUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import static com.mmc.delay.util.LockUtil.STOP_LOCK;


public class DelayFacadeImpl implements DelayFacade {


    private JobPool jobPool;

    private DelayBucket delayBucket;

    private ReadyQueue readyQueue;

    private RedisLockHelper redisLockHelper;

    private DeadLetterSet deadLetterSet;

    private static final Logger logger = LoggerFactory.getLogger(DelayFacadeImpl.class);


    public DelayFacadeImpl(RedisTemplate redisTemplate) {
        this.jobPool=new JobPool(redisTemplate);
        this.delayBucket=new DelayBucket(redisTemplate);
        this.readyQueue=new ReadyQueue(redisTemplate);
        this.deadLetterSet=new DeadLetterSet(redisTemplate);
        this.redisLockHelper=new RedisLockHelper(redisTemplate);
        initTimer();
    }
    AtomicInteger log=new AtomicInteger(0);
    /**
     * 获得最新的延期任务
     * @return  返回下一个job的runtime
     */
    @Override
    public long moveAndRtNextRunTime(String partition) {
        Set<ZSetOperations.TypedTuple> ranges = delayBucket.range(partition,DelayCoreConfig.getDelayBucketBatchSize());
        if (CollectionUtils.isEmpty(ranges)) {
            return System.currentTimeMillis()+DelayCoreConfig.getDelayNotMessageWaitTime();
        }
        for (ZSetOperations.TypedTuple member:ranges){
            Double score = member.getScore();
            if(score==null)
                return System.currentTimeMillis()+DelayCoreConfig.getDelayNotMessageWaitTime();
            //是否停止
            String topicAndIdStr = String.valueOf(member.getValue());
            TopicAndId topicAndId = new TopicAndId(topicAndIdStr);
            if(DelayCoreConfig.getStopTopics().contains(topicAndId.getTopic())){
                continue;
            }
            //可以执行了
            if(score<=System.currentTimeMillis()){
                try {
                    if(topicAndId.getTopic().equals("LOG"))
                        logger.info("移入read"+log.incrementAndGet());
                    readyQueue.addReadQueue(topicAndId.getTopicId());
                    Long remove = delayBucket.remove(partition,topicAndIdStr);
                    if(remove==0){
                        logger.info("移除delayBucket异常，topicAndId:{}",topicAndIdStr);
                    }
                } catch (Exception e) {
                    logger.error("remove data to ready error，topicAndId:{}",topicAndIdStr,e);
                }
            }
        }
        return nextScore(partition,0L);
    }


    private long nextScore(String partition,Long start) {
        Set<ZSetOperations.TypedTuple> members = delayBucket.range(partition,start,start+1);
        if (CollectionUtils.isEmpty(members)) {
            return System.currentTimeMillis()+DelayCoreConfig.getDelayNotMessageWaitTime();
        }
        ZSetOperations.TypedTuple member = members.toArray(new ZSetOperations.TypedTuple[0])[0];
        Double score = member.getScore();
        String topicAndIdStr = String.valueOf(member.getValue());
        TopicAndId topicAndId = new TopicAndId(topicAndIdStr);
        if(score==null)
            return System.currentTimeMillis()+DelayCoreConfig.getDelayNotMessageWaitTime();
        else if(DelayCoreConfig.getStopTopics().contains(topicAndId.getTopic())){
            return nextScore(partition,start+1);
        }else
            return score.longValue();
    }

    @Override
    public boolean addJob(JobParam jobParam, int currentRetryTime) {
        logger.info("添加任务：{}", jobParam.toString());
        checkAddJobParam(jobParam);
        List<SendInterceptor> sendInterceptors = DelayCoreConfig.getSendInterceptor();
        if(sendInterceptors!=null&&sendInterceptors.size()>0){
            for (SendInterceptor<JobParam> sendInterceptor : sendInterceptors) {
                jobParam = sendInterceptor.doSend(jobParam);
            }
        }
        try{
            Job job=new Job();
            BeanUtils.copyProperties(jobParam,job);
            long currentTimeMillis = System.currentTimeMillis();
            if(jobParam.getDelayTime()>0){
                job.setShoudRunTime(currentTimeMillis+jobParam.getDelayTime());
            }
            job.setCurrentRetryCount(currentRetryTime+1);
            jobPool.addJob(job);
            if(job.getShoudRunTime()>currentTimeMillis){
                String partition = new RandomPartitionService().partition(job.getTopic());
                delayBucket.addDelayJob(partition,job);
                LockUtil.tryUpdateTime(partition,job.getShoudRunTime());
            }else {
                readyQueue.addReadQueue(job.getJobId());
            }

        }catch (Exception e){
            logger.error("添加任务异常",e);
            return false;
        }
        return true;
    }

    /**
     * 初始化定时器
     */
    private void initTimer() {
        if (checkIsNeedSingleNodeRun()) return;
        new DelayMoveToReadyTimer(this).runMove();
        new ReadyQueueTimer(this).runConsume();
    }

    private boolean checkIsNeedSingleNodeRun() {
        String timerIp = DelayCoreConfig.getTimerIp();
        if (!StringUtils.isEmpty(timerIp)) {
            try {
                String hostAddress = InetAddress.getLocalHost().getHostAddress();
                //不是指定的ip就不运行定时任务
                if (!hostAddress.equals(timerIp)) {
                    return true;
                }
            } catch (UnknownHostException e) {
                logger.error("获取本机ip异常", e);
            }
        }
        return false;
    }

    @Override
    public boolean addJob(JobParam jobParam) {
        return addJob(jobParam,0);
    }

    private void checkAddJobParam(JobParam jobParam) {
        if(jobParam.getJobId()==null){
            long nextId = SnowFlakeShortUtil.getInstance().nextId();
            jobParam.setJobId(String.valueOf(nextId));
        }
        if(StringUtils.isEmpty(jobParam.getTopic())){
            throw new DelayException("topic不能为空");
        }
        if(jobParam.getTopic().contains(TopicAndId.COLON)){
            throw new DelayException("topic不能含有冒号");
        }
        if(DelayCoreConfig.getJobHandle(jobParam.getTopic())==null){
            throw new DelayException("请先调用producer.setJobHandle方法添加消息处理逻辑");
        }
    }


    @Override
    public List<Job> popReadyJob() {
        List<String> jobIds = readyQueue.popJob();
        if(CollectionUtils.isEmpty(jobIds)){
            return Collections.EMPTY_LIST;
        }
        List<Job> jobs = jobPool.mget(jobIds);
        return jobs;
    }

    @Override
    public boolean removeJob(String jobId) {
        jobPool.removeJob(jobId);
        return true;
    }

    @Override
    public boolean tryRetry(Job job) {
        if(job.getCurrentRetryCount()<job.getRetries()){
            JobParam jobParam=new JobParam();
            BeanUtils.copyProperties(job,jobParam);
            addJob(jobParam,job.getCurrentRetryCount());
        }else {
            logger.error("重试次数以达上限，请手工处理,jobId:{}",job.getJobId());
            DeadJob deadJob=new DeadJob();
            BeanUtils.copyProperties(job,deadJob);
            deadJob.setCreateTime(System.currentTimeMillis());
            deadLetterSet.addDeadLetterQueue(deadJob);
            removeJob(job.getJobId());
            JobHandle jobHandle = DelayCoreConfig.getJobHandle(job.getTopic());
            jobHandle.errorHandle(job);
        }
        return true;
    }


    @Override
    public boolean lockTimer(String key,String value,Long second) {
        return redisLockHelper.lock(key,value,second);
    }

    @Override
    public boolean unlock(String key,String value) {
        return redisLockHelper.unlock(key,value);
    }

    @Override
    public List<Job> queryDelayData(String topic,Long start,Long end) {
        List<Job> result=new ArrayList<>();
        Set<ZSetOperations.TypedTuple> range = delayBucket.range(topic, start, end);
        if(!CollectionUtils.isEmpty(range)){
            for (ZSetOperations.TypedTuple r:range){
                Job job = jobPool.getJob(new TopicAndId(r.getValue().toString()).getTopicId());
                result.add(job);
            }
        }
        return result;
    }

    @Override
    public Long queryDelayRecordNum(String topic) {
        return delayBucket.totalNum(topic);
    }

    @Override
    public List<DeadJob> queryDeadData(String topic, Long start, Long end) {
        List<DeadJob> result=new ArrayList<>();
        Set<ZSetOperations.TypedTuple> range = deadLetterSet.range(topic, start, end);
        if(CollectionUtils.isEmpty(range)){
            return result;
        }
        for (ZSetOperations.TypedTuple r:range){
            Object value = r.getValue();
            result.add((DeadJob)value);
        }
        return result;
    }

    @Override
    public Long queryDeadRecordNum(String topic) {
        return deadLetterSet.totalNum(topic);
    }

    @Override
    public void removeDeadJob(DeadJob deadJob) {
        deadLetterSet.removeDeadJob(deadJob);
    }

    @Override
    public Set<String> queryAllTopics() {
        return DelayCoreConfig.topicList();
    }

    @Override
    public Set<String> queryStopTopics() {
        return DelayCoreConfig.getStopTopics();
    }

    @Override
    public void stopTopic(String topic) {
        if(StringUtils.isEmpty(topic)){
            throw new DelayException("topic不能为空");
        }
        DelayCoreConfig.addStopTopic(topic);
    }

    @Override
    public void startTopic(String topic) {
        if(StringUtils.isEmpty(topic)){
            throw new DelayException("topic不能为空");
        }
        DelayCoreConfig.removeStopTopic(topic);
        synchronized (STOP_LOCK){
            STOP_LOCK.notifyAll();
        }
    }


}
